﻿//IQFeedWrapper.cs
//Copyright (c) 2013 StockSharp LLC, all rights reserved.
//This code module is part of StockSharp library.
//This code is licensed under the GNU GENERAL PUBLIC LICENSE Version 3.
//See the file License.txt for the license details.
//More info on: http://stocksharp.com

namespace StockSharp.Algo.History.IQFeed
{
	using System;
	using System.Collections.Generic;
	using System.Globalization;
	using System.Linq;
	using System.Net;
	using System.Net.Sockets;
	using System.Text;
	using System.Text.RegularExpressions;
	using System.Threading;

	using Ecng.Collections;
	using Ecng.Common;
	using Ecng.Net;

	/// <summary>
	/// Обертка для работы с IQFeed c помощью протокола TCP/IP.
	/// </summary>
	class IQFeedWrapper
	{
		enum IQFeedCommandType
		{
			Connect,
			Disconnect,
			IqFeed,
		}

		private sealed class IQFeedCommand
		{
			public IQFeedCommandType Type { get; private set; }

			public Action Action { get; private set; }

			public IQFeedCommand(IQFeedCommandType type, Action action)
			{
				Type = type;
				Action = action;
			}
		}

		private readonly Regex _regexError = new Regex("^E,(.*?),*?$", RegexOptions.Compiled);
		private readonly Regex _regexTime = new Regex("^T,(.*?)$", RegexOptions.Compiled);
		private readonly Regex _regexLastMessage = new Regex("^!ENDMSG!,$", RegexOptions.Compiled);
		private readonly Regex _regexRequestId = new Regex("^(#.*?#),(.*?)$", RegexOptions.Compiled);
		private readonly Regex _regex = new Regex("^(S|n|Z|2|P|Q),([a-zA-Z].*?)$", RegexOptions.Compiled);

		private readonly Dictionary<string, IQFeedMessageType> _requestsType = new Dictionary<string, IQFeedMessageType>();
		private readonly object _syncDisconnect = new object();
		private readonly IdGenerator _requestId = new IdGenerator();
		private Socket _socket;
		private IPEndPoint _address = new IPEndPoint(IPAddress.Loopback, 9100);

		private readonly BlockingQueue<IQFeedCommand> _requests = new BlockingQueue<IQFeedCommand>();

		/// <summary>
		/// Адрес сервера.
		/// </summary>
		public IPEndPoint Address
		{
			get { return _address; }
			set
			{
				if (value == null)
					throw new ArgumentNullException("value");

				_address = value;
			}
		}

		/// <summary>
		/// Запущен ли экспорт. Экспорт запускается при отправке IQFeed команды./>.
		/// </summary>
		public bool IsExportStarted { get; private set; }

		/// <summary>
		/// Событие ошибки подключения (например, соединение было разорвано).
		/// </summary>
		public event Action<Exception> ConnectionError;

		/// <summary>
		/// Событие, сигнализирующее об ошибке при обработке новых данных с сервера.
		/// </summary>
		public event Action<Exception> ProcessDataError;

		/// <summary>
		/// Событие появления новых данных.
		/// </summary>
		public event Action<IEnumerable<IQFeedMessage>> ProcessReply;

		public IQFeedWrapper()
		{
			ThreadingHelper
				.Thread(() =>
				{
					IQFeedCommand command;

					while (_requests.TryDequeue(out command))
					{
						try
						{
							var needStart = true;

							if (command.Type == IQFeedCommandType.Disconnect)
							{
								needStart = false;
								StopExport();
							}

							command.Action.SafeInvoke();

							if (needStart && !IsExportStarted)
								StartExport();
						}
						catch (Exception ex)
						{
							ConnectionError.SafeInvoke(ex);
						}
					}
				})
				.Name("IQFeed request thread")
				.Culture(CultureInfo.InvariantCulture)
				.Start();
		}

		private void StartExport()
		{
			IsExportStarted = true;

			ThreadingHelper
				.Thread(() =>
				{
					var buf = new StringBuilder();
					var buffer = new byte[1024];

					while (true)
					{
						try
						{
							if (_socket == null || !_socket.IsConnected())
								throw new InvalidOperationException("Соединение не было установлено.");

							if (_socket.Wait(500000))
							{
								var bytesRecv = _socket.Receive(buffer, 0, buffer.Length, SocketFlags.None);

								buf.Append(Encoding.ASCII.GetString(buffer, 0, bytesRecv));

								var index = buf.LastIndexOf('\n');

								if (index == -1)
									continue;

								var res = buf.ToString(0, index + 1);
								var reply = ParseReply(res);

								if (index + 1 < buf.Length - 1)
									buf.Remove(0, index + 1);
								else
									buf.Clear();

								ProcessReply.SafeInvoke(reply);
							}
						}
						catch (ObjectDisposedException ex)
						{
							ConnectionError.SafeInvoke(new InvalidOperationException("Соединение было закрыто.", ex));
							break;
						}
						catch (Exception ex)
						{
							ConnectionError.SafeInvoke(ex);
							break;
						}

						lock (_syncDisconnect)
						{
							if (IsExportStarted)
								continue;

							Monitor.Pulse(_syncDisconnect);
							break;
						}
					}
				})
				.Name("IQFeed response thread")
				.Culture(CultureInfo.InvariantCulture)
				.Start();
		}

		private void StopExport()
		{
			lock (_syncDisconnect)
			{
				IsExportStarted = false;
				Monitor.Wait(_syncDisconnect);
			}
		}

		/// <summary>
		/// Подключиться к серверу.
		/// </summary>
		public void Connect()
		{
			_requests.Enqueue(new IQFeedCommand(IQFeedCommandType.Connect, () =>
			{
				if (_socket != null)
					throw new InvalidOperationException("Соединение не было остановлено с предыдущего подключения.");

				_socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);

				try
				{
					_socket.Connect(Address);
				}
				catch
				{
					_socket.Close();
					_socket = null;
					throw;
				}	
			}));

			Request("S,SET PROTOCOL,5.0\r\n");
		}

		/// <summary>
		/// Отключиться от сервера.
		/// </summary>
		public void Disconnect()
		{
			_requests.Enqueue(new IQFeedCommand(IQFeedCommandType.Disconnect, () =>
			{
				if (_socket == null)
					return;

				_socket.Shutdown(SocketShutdown.Both);
				_socket.Close();
				_socket = null;
			}));
		}

		private void Send(string command)
		{
			if (_socket == null || !_socket.IsConnected())
				throw new InvalidOperationException("Соединение не было установлено.");

			var request = Encoding.ASCII.GetBytes(command);

			var bytesSent = _socket.Send(request, request.Length, SocketFlags.None);

			if (bytesSent != request.Length)
				throw new InvalidOperationException("Ошибка при отправке команды: {0}".Put(command.TrimEnd("\r\n".ToCharArray())));
		}

		private string GetRequestId()
		{
			return "#" + _requestId.Next + "#";
		}

		private IEnumerable<IQFeedMessage> ParseReply(string reply)
		{
			var result = new List<IQFeedMessage>();
			var lines = reply.Split(new[] { '\n' }, StringSplitOptions.RemoveEmptyEntries);

			foreach (var tempLine in lines.Select(v => v.TrimEnd('\r', '\n')))
			{
				var requestId = String.Empty;
				var line = tempLine;

				var match = _regexRequestId.Match(line);
				if (match.Success)
				{
					requestId = match.Groups[1].Value;
					line = match.Groups[2].Value;
				}

				try
				{
					match = _regexError.Match(line);
					if (match.Success)
					{
						var text = match.Groups[1].Value;
						if (!text.Equals("!NO_DATA!"))
							result.Add(text.ToMessage(IQFeedMessageType.Error));

						continue;
					}

					match = _regexTime.Match(line);
					if (match.Success)
					{
						result.Add(match.Groups[1].Value.ToMessage(IQFeedMessageType.Time));
						continue;
					}

					match = _regexLastMessage.Match(line);
					if (match.Success)
					{
						_requestsType.Remove(requestId);
						result.Add(match.Groups[1].Value.ToMessage(IQFeedMessageType.End));
						continue;
					}

					match = _regex.Match(line);
					if (match.Success)
					{
						var cmd = match.Groups[1].Value;
						switch (cmd)
						{
							case "S":
								result.Add(match.Groups[2].Value.ToMessage(IQFeedMessageType.System));
								continue;
							case "n":
								result.Add("Инструмент {0} не найден.".Put(match.Groups[2].Value).ToMessage(IQFeedMessageType.Error));
								continue;
							case "Z":
							case "2":
								//временное решение, т.к. с "2" может начинаться не только обновление стакана.
								if (_requestsType.Count == 0)
								{
									result.Add(match.Groups[2].Value.ToMessage(IQFeedMessageType.Quote));
									continue;
								}
								break;
							case "P":
							case "Q":
								//TODO: esper, необходимо добавить поддержку пользовательского набора полей
								result.Add(match.Groups[2].Value.ToMessage(IQFeedMessageType.Trade));
								continue;
						}
					}

					IQFeedMessageType type;

					if (!_requestsType.TryGetValue(requestId, out type))
						type = IQFeedMessageType.Data;

					result.Add(line.ToMessage(type));
				}
				catch (Exception ex)
				{
					ProcessDataError.SafeInvoke(ex);
				}
			}

			return result;
		}

		/// <summary>
		/// Отправить запрос на получение данных.
		/// </summary>
		/// <param name="command">Запрос в формате IQFeed.</param>
		public void Request(string command)
		{
			_requests.Enqueue(new IQFeedCommand(IQFeedCommandType.IqFeed, () => Send(command)));
		}

		/// <summary>
		/// Отправить запрос на получение списка торговых площадок.
		/// </summary>
		public void RequestListedMarkets()
		{
			//Request a list of Listed Markets from the feed. 
			//SLM<CR><LF> 
			//Returns: A list of records. Each record identifies a single listed market in the following format: 
			//[Listed Market ID],[Short Name],[Long Name],<LF>

			_requestsType.Add("", IQFeedMessageType.ListedMarket);

			Request("SLM\r\n");
		}

		/// <summary>
		/// Отправить запрос на получение списка инструментов по заданному фильтру.
		/// </summary>
		/// <param name="searchField">Поле, по которому необходимо искать данные.</param>
		/// <param name="searchText">Строка поиска.</param>
		/// <param name="filterType">Тип фильтра.</param>
		/// <param name="filterValue">Значения фильтра.</param>
		public void RequestSecurities(IQFeedSearchField searchField, string searchText, IQFeedFilterType filterType, params string[] filterValue)
		{
			//Symbols By Filter - A symbol search by symbol or description. Can be filtered by providing a list of Listed Markets or Security Types. 
			//SBF,[Field To Search],[Search String],[Filter Type],[Filter Value],[RequestID]<CR><LF> 
			//[Field To Search] - "s" to search symbols. "d" to search descriptions.
			//[Search String] - What you want to search for.
			//[Filter Type] - "e" to search within specific Listed Markets. "t" to search within specific Security Types.
			//[Filter Value] - A space delimited list of listed markets or security types (based upon the Filter Type parameter).
			//NOTE: A list of security types or listed marekts can be retrieved dynamicly from the server using the SLM and SST requests below.
			//[RequestID] - This parameter allows you to specify an identier that will be attached to the returned data inserted as the FIRST field in the output message.

			var id = GetRequestId();
			var request = "SBF,{0},{1},{2},{3},{4}\r\n".Put((searchField == IQFeedSearchField.Symbol) ? "s" : "d",
				searchText, (filterType == IQFeedFilterType.Market) ? "e" : "t", filterValue.Join(" "), id);

			_requestsType.Add(id, IQFeedMessageType.Security);

			Request(request);
		}

		/// <summary>
		/// Отправить запрос на получение истории сделок по инструменту за указанный период.
		/// </summary>
		/// <param name="symbol">Код инструмента.</param>
		/// <param name="from">Начало периода.</param>
		/// <param name="to">Конец периода.</param>
		public void RequestTrades(string symbol, DateTime from, DateTime to)
		{
			//Retrieves tick data between [BeginDate BeginTime] and [EndDate EndTime] for the specified [Symbol].
			//HTT,[Symbol],[BeginDate BeginTime],[EndDate EndTime],[MaxDatapoints],[BeginFilterTime],[EndFilterTime],[DataDirection],[RequestID],[DatapointsPerSend]<CR><LF> 
			//[Symbol] - Required - Max Length 30 characters.
			//[BeginDate BeginTime] - Required if [EndDate EndTime] not specified - Format CCYYMMDD HHmmSS - Earliest date/time to receive data for.
			//[EndDate EndTime] - Required if [BeginDate BeginTime] not specified - Format CCYYMMDD HHmmSS - Most recent date/time to receive data for.
			//[MaxDatapoints] - Optional - the maximum number of datapoints to be retrieved.
			//[BeginFilterTime] - Optional - Format HHmmSS - Allows you to specify the earliest time of day (Eastern) for which to receive data.
			//[EndFilterTime] - Optional - Format HHmmSS - Allows you to specify the latest time of day (Eastern) for which to receive data.
			//[DataDirection] - Optional - '0' (default) for "newest to oldest" or '1' for "oldest to newest".
			//[RequestID] - Optional - Will be sent back at the start of each line of data returned for this request.
			//[DatapointsPerSend] - Optional - Specifies the number of datapoints that IQConnect.exe will queue before attempting to send across the socket to your app.

			var id = GetRequestId();
			var request = "HTT,{0},{1},{2},,000000,235959,1,{3}\r\n".Put(symbol, from.ToString("yyyyMMdd HHmmss"), to.ToString("yyyyMMdd HHmmss"), id);

			_requestsType.Add(id, IQFeedMessageType.HistoryTrade);

			Request(request);
		}

		/// <summary>
		/// Подписаться на получение данных по инструменту.
		/// </summary>
		/// <param name="symbol">Код инструмента.</param>
		public void SubscribeSymbol(string symbol)
		{
			Request("w{0}\r\n".Put(symbol));
		}

		/// <summary>
		/// Отписаться от получения данных по инструменту.
		/// </summary>
		/// <param name="symbol">Код инструмента.</param>
		public void UnSubscribeSymbol(string symbol)
		{
			Request("r{0}\r\n".Put(symbol));
		}

		/// <summary>
		/// Установить набор полей для получения данных по Level1.
		/// </summary>
		/// <param name="fields">Названия полей.</param>
		public void SetLevel1FieldSet(params string[] fields)
		{
			Request("S,SELECT UPDATE FIELDS,{0}\r\n".Put(fields.Join(",")));
		}

		/// <summary>
		/// Установить набор полей для Level1 сообщения для получения данных по сделкам.
		/// </summary>
		public void SetDefaultLevel1FieldSet()
		{
			SetLevel1FieldSet("TickID", "Last Trade Date", "Most Recent Trade TimeMS", "Most Recent Trade", "Most Recent Trade Size", "Message Contents");
		}
	}
}
